// +build aio

package net

import (
	"container/list"
	"errors"
	//"fmt"
	"github.com/sniperHW/fly/core/buffer"
	"github.com/sniperHW/goaio"
	"math/rand"
	"net"
	"runtime"
	"sync"
	"time"
)

type BufferPool struct {
	pool sync.Pool
}

const PoolBuffSize uint64 = 1024 * 1024 * 2

var buffPool *BufferPool = NewBufferPool()

var aioService *SocketService = NewSocketService(buffPool)

func NewBufferPool() *BufferPool {
	return &BufferPool{
		pool: sync.Pool{
			New: func() interface{} {
				return make([]byte, PoolBuffSize)
			},
		},
	}
}

func (p *BufferPool) Acquire() []byte {
	return p.pool.Get().([]byte)
}

func (p *BufferPool) Release(buff []byte) {
	if uint64(cap(buff)) == PoolBuffSize {
		p.pool.Put(buff[:cap(buff)])
	}
}

func GetBuffPool() *BufferPool {
	return buffPool
}

type SocketService struct {
	services          []*goaio.AIOService
	outboundTaskQueue []*goaio.TaskQueue
	shareBuffer       goaio.ShareBuffer
}

type ioContext struct {
	s *Socket
	t rune
}

func (this *SocketService) completeRoutine(s *goaio.AIOService) {
	for {
		res, err := s.GetCompleteStatus()
		if nil != err {
			break
		} else {
			context := res.Context.(*ioContext)
			if context.t == 'r' {
				context.s.onRecvComplete(&res)
			} else {
				context.s.onSendComplete(&res)
			}
		}
	}
}

func (this *SocketService) bind(conn net.Conn) (*goaio.AIOConn, *goaio.TaskQueue, error) {
	idx := rand.Int() % len(this.services)
	c, err := this.services[idx].Bind(conn, goaio.AIOConnOption{
		SendqueSize: 1,
		RecvqueSize: 1,
		ShareBuff:   this.shareBuffer,
	})
	return c, this.outboundTaskQueue[idx], err
}

func (this *SocketService) outboundRoutine(tq *goaio.TaskQueue) {
	for {
		var err error
		queue := make([]interface{}, 0, 512)
		for {
			queue, err = tq.Pop(queue)
			if nil != err {
				return
			} else {
				for _, v := range queue {
					v.(*Socket).doSend()
				}
			}
		}
	}
}

func (this *SocketService) Close() {
	for i, _ := range this.services {
		this.services[i].Close()
		this.outboundTaskQueue[i].Close()
	}
}

func NewSocketService(shareBuffer goaio.ShareBuffer) *SocketService {
	s := &SocketService{
		shareBuffer: shareBuffer,
	}

	for i := 0; i < 2; i++ {
		se := goaio.NewAIOService(2)
		tq := goaio.NewTaskQueue()
		s.services = append(s.services, se)
		s.outboundTaskQueue = append(s.outboundTaskQueue, tq)
		go s.completeRoutine(se)
		go s.outboundRoutine(tq)
	}

	return s
}

type Socket struct {
	socketBase
	muW           sync.Mutex
	sendQueue     *list.List
	aioConn       *goaio.AIOConn
	sendQueueSize int
	sendLock      bool
	tq            *goaio.TaskQueue
	sendContext   ioContext
	recvContext   ioContext
	b             *buffer.Buffer
	offset        int
}

func (s *Socket) SetSendQueueSize(size int) *Socket {
	s.muW.Lock()
	defer s.muW.Unlock()
	s.sendQueueSize = size
	return s
}

func (s *Socket) SetRecvTimeout(timeout time.Duration) *Socket {
	s.aioConn.SetRecvTimeout(timeout)
	return s
}

func (s *Socket) SetSendTimeout(timeout time.Duration) *Socket {
	s.aioConn.SetSendTimeout(timeout)
	return s
}

func (s *Socket) onRecvComplete(r *goaio.AIOResult) {
	if s.testFlag(fclosed | frclosed) {
		s.ioWait.Done()
	} else {
		recvAgain := false

		defer func() {
			if !s.testFlag(fclosed|frclosed) && recvAgain {
				b := s.inboundProcessor.GetRecvBuff()
				if nil != s.aioConn.Recv(b, &s.recvContext) {
					s.ioWait.Done()
				}
			} else {
				s.ioWait.Done()
			}
		}()

		if nil != r.Err {

			if r.Err == goaio.ErrRecvTimeout {
				r.Err = ErrRecvTimeout
			}

			if nil != s.errorCallback {
				if r.Err == ErrRecvTimeout {
					recvAgain = true
				} else {
					this.setFlag(frclosed)
				}
				s.errorCallback(s, r.Err)
			} else {
				s.Close(r.Err, 0)
			}

		} else {
			s.inboundProcessor.OnData(r.Buff[:r.Bytestransfer])
			for !s.testFlag(fclosed | frclosed) {
				msg, err := s.inboundProcessor.Unpack()
				if nil != err {
					s.Close(err, 0)
					if nil != s.errorCallback {
						s.errorCallback(s, err)
					}
					break
				} else if nil != msg {
					s.inboundCallBack(s, msg)
				} else {
					recvAgain = true
					break
				}
			}
		}
	}
}

func (s *Socket) emitSendTask() {
	s.ioWait.Add(1)
	if nil == s.tq.Push(s) {
		s.sendLock = true
	} else {
		s.ioWait.Done()
		s.sendLock = false
	}
}

func (s *Socket) doSend() {
	const maxsendsize = 65535
	var err error

	s.muW.Lock()
	//只有之前请求的buff全部发送完毕才填充新的buff
	if nil == s.b {
		s.b = buffer.Get()
		for v := s.sendQueue.Front(); v != nil; v = s.sendQueue.Front() {
			s.sendQueue.Remove(v)
			if err = s.encoder.EnCode(v.Value, s.b); nil != err {
				s.b.Free()
				s.b = nil
				break
			} else if s.b.Len() >= maxsendsize {
				break
			}
		}
		s.offset = 0
	}

	s.muW.Unlock()

	if nil == err {
		if nil != s.aioConn.Send(s.b.Bytes()[s.offset:], &s.sendContext) {
			s.ioWait.Done()
		}
	} else {
		s.ioWait.Done()
		if !s.testFlag(fclosed) {
			if nil != s.errorCallback {
				s.Close(err, 0)
				s.errorCallback(s, err)
			} else {
				s.Close(err, 0)
			}
		}
	}
}

func (s *Socket) onSendComplete(r *goaio.AIOResult) {
	defer s.ioWait.Done()
	if nil == r.Err {
		s.muW.Lock()
		defer s.muW.Unlock()
		//发送完成释放发送buff
		s.b.Free()
		s.b = nil
		if s.sendQueue.Len() == 0 {
			s.sendLock = false
			if s.testFlag(fclosed) {
				close(s.sendCloseChan)
			}
		} else {
			s.emitSendTask()
		}
	} else if !s.testFlag(fclosed) {
		if r.Err == goaio.ErrSendTimeout {
			r.Err = ErrSendTimeout
			//timeout也可能会完成部分字节的发送
			s.offset += r.Bytestransfer
		}

		if nil != s.errorCallback {
			if r.Err != ErrSendTimeout {
				s.Close(r.Err, 0)
			}
			s.errorCallback(s, r.Err)

			//发送超时，但用户没有关闭socket,需要将尚未发送完成的buff再次请求发送
			if r.Err == ErrSendTimeout && !s.testFlag(fclosed) {
				s.muW.Lock()
				s.emitSendTask()
				s.muW.Unlock()
			}

		} else {
			s.Close(r.Err, 0)
		}
	}
}

func (s *Socket) Send(o interface{}) error {
	if s.testFlag(fclosed) {
		return ErrSocketClose
	}

	if s.testFlag(fclosed) {
		return ErrSocketClose
	}

	s.muW.Lock()
	defer s.muW.Unlock()

	if s.sendQueue.Len() > s.sendQueueSize {
		return ErrSendQueFull
	}

	s.sendQueue.PushBack(o)

	if !s.sendLock {
		s.emitSendTask()
	}

	return nil

}

func (s *Socket) BeginRecv(cb func(*Socket, interface{})) (err error) {
	s.beginOnce.Do(func() {
		if nil == cb {
			panic("BeginRecv cb is nil")
		}

		if nil == s.inboundProcessor {
			panic("inboundProcessor is nil")
		}

		if s.testFlag(fclosed | frclosed) {
			err = ErrSocketClose
		} else {
			//发起第一个recv
			s.inboundCallBack = cb

			s.ioWait.Add(1)
			if err = s.aioConn.Recv(s.inboundProcessor.GetRecvBuff(), &s.recvContext); nil != err {
				s.ioWait.Done()
			}

		}
	})
	return
}

func (s *Socket) Close(reason error, delay time.Duration) {
	s.closeOnce.Do(func() {
		runtime.SetFinalizer(s, nil)

		s.setFlag(fclosed)

		s.muW.Lock()
		if s.sendQueue.Len() > 0 {
			delay = delay * time.Second
		} else {
			delay = 0
		}
		s.muW.Unlock()

		if delay > 0 {
			s.ShutdownRead()
			ticker := time.NewTicker(delay)
			go func() {
				select {
				case <-s.sendCloseChan:
				case <-ticker.C:
				}

				ticker.Stop()
				s.aioConn.Close(nil)
			}()
		} else {
			s.aioConn.Close(nil)
		}

		go func() {
			s.ioWait.Wait()
			if nil != s.b {
				s.b.Free()
			}
			if nil != s.inboundProcessor {
				s.inboundProcessor.OnSocketClose()
			}
			if nil != s.closeCallBack {
				s.closeCallBack(s, reason)
			}
		}()
	})
}

func NewSocket(service *SocketService, conn net.Conn) *Socket {
	switch conn.(type) {
	case *net.TCPConn, *net.UnixConn:
		break
	default:
		return nil
	}

	s := &Socket{
		socketBase: socketBase{
			conn:          conn,
			sendCloseChan: make(chan struct{}),
		},
	}

	c, tq, err := service.bind(conn)
	if err != nil {
		return nil
	}

	s.tq = tq
	s.aioConn = c
	s.sendQueueSize = 256
	s.sendQueue = list.New()
	s.sendContext = ioContext{s: s, t: 's'}
	s.recvContext = ioContext{s: s, t: 'r'}

	runtime.SetFinalizer(s, func(s *Socket) {
		s.Close(errors.New("gc"), 0)
	})

	return s
}

func CreateSocket(conn net.Conn) *Socket {
	return NewSocket(aioService, conn)
}
